package com.bbcare.mqservice.utils.rockermq;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;


import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class ConsumerFactory {
	private static  final Logger logger = LogManager.getLogger();
    @Autowired
	 private RocketmqListener rocketmqListener;
		

	@Value("${aliyun.mq.access_id}")
    private String ACCESS_ID;
	@Value("${aliyun.mq.access_key}")
	private String SecretKey;
	@Value("${aliyun.mq.access_expired_time}")
	private String ACCESS_EXPIRED_TIME;
	@Value("${aliyun.mq.namesrv_addr}")
	private String NAMESRV_ADDR;
	@Value("${aliyun.mq.receive.msgTopic}")
	private String msgReceiveTopic;
	@Value("${aliyun.mq.group_consumer_id}")
	private String GROUP_ID;

	public static final String tag = "*";

	//消费者实例配置列表

	private static Consumer consumer;
	

	
	/**
	 * 消费者启动
	 * @throws Exception 
	 */
	public void start() {
		    String[] Topics = msgReceiveTopic.split(",");
			Properties prop = new Properties();
			prop.put(PropertyKeyConst.AccessKey,ACCESS_ID);// AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
			prop.put(PropertyKeyConst.SecretKey,SecretKey);// SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
		    prop.put(PropertyKeyConst.GROUP_ID,GROUP_ID);//您在控制台创建的Producer ID
		    prop.put(PropertyKeyConst.SendMsgTimeoutMillis,ACCESS_EXPIRED_TIME);//设置发送超时时间，单位毫秒
			prop.put(PropertyKeyConst.NAMESRV_ADDR,NAMESRV_ADDR);
			consumer = ONSFactory.createConsumer(prop);
			for(String topic : Topics){
			consumer.subscribe(topic, tag,  rocketmqListener);//监听第一个topic，new对应的监听器
		}

		// 在发送消息前，必须调用start方法来启动consumer，只需调用一次即可，当项目关闭时，自动shutdown
		consumer.start();
		logger.info("ConsumerConfig start success.");

	}
	
	/**
	 * 停消费者
	 */
	public void shutdown(){
		if(consumer!=null)
		consumer.shutdown();
	}


	public  Consumer getConsumer()
	{
		if (consumer == null)
		{
			synchronized(Consumer.class) {
				if (consumer == null)
					start();
			}
		}
		return consumer;
	}

	

}
